// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2025-02-08

package binlog

import (
	"context"
	"encoding/json"
	"fmt"
	"gitee.com/go-libs/log"
	"gitee.com/go-libs/result"
	"gitee.com/go-libs/validator-iris/validator"
	"gitee.com/go-wares/framework-iris/framework/src/logics"
	"github.com/kataras/iris/v12"
	"github.com/valyala/fasthttp"
	"integration/app/lib/conf"
	"net/http"
)

// Logic
// 处理Binlog.
type Logic struct {
	request  *Request
	response *Response
}

// New
// 创建实例.
func New() logics.Logical {
	return &Logic{
		request:  &Request{},
		response: &Response{},
	}
}

// Before
// 前置校验.
func (o *Logic) Before(_ context.Context, i iris.Context) (err error) {
	if err = validator.Json(i, o.request); err == nil {
		o.response.Database = o.request.Database
		o.response.Table = o.request.Table
	}
	return
}

// Run
// 执行逻辑.
func (o *Logic) Run(ctx context.Context, _ iris.Context) result.Result {
	var (
		bin   *conf.ErpBinlog
		has   bool
		res   = result.New()
		table = o.request.TableKey()
	)

	// 1. 获取配置.
	//    指定的数据表在 erp.yaml 配置文件中未配置.
	//
	//   # config/erp.yaml
	//   binlog:
	//     database.table:
	//       topic: "order.status"
	//       tag: "tag"
	if bin, has = conf.Config.GetBinlog(table); !has {
		return res.With(map[string]any{
			"table":   table,
			"message": "not configured in erp.yaml",
		})
	}

	// 2. 遍历消息.
	for _, data := range o.request.Data {
		o.publish(ctx, bin, data)
	}
	return res.With(o.response)
}

func (o *Logic) publish(cts context.Context, bin *conf.ErpBinlog, data map[string]any) {
	var (
		tag     = bin.GenerateTag(data)
		payload = map[string]any{
			"topic": bin.Topic, "tag": tag, "message": map[string]any{
				"data":     data,
				"database": o.request.Database,
				"table":    o.request.Table,
				"type":     o.request.Type,
			},
		}
		body, _ = json.Marshal(payload)

		ctx = log.NewContextForChildInfo(cts, `[mq] publish message: topic="%s", tag="%s"`, bin.Topic, tag)
		err error
		req = fasthttp.AcquireRequest()
		res = fasthttp.AcquireResponse()
	)

	o.response.Messages++

	// 1. 监听结束.
	defer func() {
		// 1.1 捕获异常.
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[mq] publish message panic: fatal="%v"`, r)

			// 错误原因.
			if err == nil {
				err = fmt.Errorf(`%v`, r)
			}
		}

		// 1.2 释放资源.
		fasthttp.ReleaseRequest(req)
		fasthttp.ReleaseResponse(res)

		// 1.3 记录结果.
		if err != nil {
			log.Errorfc(ctx, `[mq] publish message error: err="%v"`, err)
		}
	}()

	// 2. 请求参数.
	req.Header.SetRequestURI(conf.Config.Mbs.Producer)
	req.Header.SetContentType("application/json")
	req.Header.SetMethod(http.MethodPost)

	// 3. 调用链路.
	for k, v := range log.ContextHeader(ctx) {
		if len(v) > 0 && v[0] != "" {
			req.Header.Set(k, v[0])
		}
	}

	// 4. 消息内容.
	req.SetBodyRaw(body)

	// 5. 结果校验.
	if err = fasthttp.Do(req, res); err == nil {
		if status := res.StatusCode(); status != http.StatusOK {
			err = fmt.Errorf(`HTTP %d %s`, status, http.StatusText(status))
			return
		}
	}

	// 6. 发布结果.
	o.response.Success++
	log.Infofc(ctx, `[mq] publish message success: %s`, res.Body())
}
